package top.bulk.sse.service.impl;

import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.IdUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import top.bulk.sse.common.Res;
import top.bulk.sse.dto.CommonSseParam;
import top.bulk.sse.service.SseEmitterService;
import top.bulk.sse.util.SseCacheUtil;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * 实现层
 *
 * @author 散装java
 * @date 2024-09-10
 */
@Slf4j
@Service
public class SseEmitterServiceImpl implements SseEmitterService {

    /**
     * 心跳线程池
     */
    private static final ScheduledExecutorService HEARTBEAT_EXECUTORS = Executors.newScheduledThreadPool(10);


    /**
     * 创建连接
     *
     * @param clientId     客户端id
     * @param loginId      登录人
     * @param setHeartBeat 是否需要心跳检测
     * @return sseEmitter
     */
    @Override
    public SseEmitter createSseConnect(String clientId, String loginId, Boolean setHeartBeat) {
        // 设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 判断连接是否有效
        if (SseCacheUtil.connectionValidity(clientId, loginId)) {
            return SseCacheUtil.getSseEmitterByClientId(clientId);
        } else {
            SseCacheUtil.removeConnection(clientId);
        }
        clientId = IdUtil.simpleUUID();
        String finalClientId = clientId;
        // 增加心跳
        ScheduledFuture<?> future = null;
        // 是否自定义心跳任务
        if (BooleanUtil.isTrue(setHeartBeat)) {
            // 设置心跳任务
            future = HEARTBEAT_EXECUTORS.scheduleAtFixedRate(() ->
                            SseCacheUtil.sendMessageToOneClient(finalClientId, finalClientId + "-" + loginId),
                    2, 10, TimeUnit.SECONDS);


        }

        // 增加连接
        SseCacheUtil.addConnection(clientId, loginId, sseEmitter, future);

        // 长链接完成后回调(即关闭连接时调用)
        sseEmitter.onCompletion(SseCacheUtil.completionCallBack(clientId));
        // 连接超时回调
        sseEmitter.onTimeout(SseCacheUtil.timeoutCallBack(clientId));
        // 推送消息异常回调
        sseEmitter.onError(SseCacheUtil.errorCallBack(clientId));
        // 初次建立连接,推送客户端id
        SseCacheUtil.sendMessageToClientByClientId(clientId, Res.ok(clientId));
        return sseEmitter;
    }

    /**
     * 关闭连接
     **/
    @Override
    public void closeSseConnect(String clientId) {
        SseCacheUtil.removeConnection(clientId);
    }

    /**
     * 推送消息到所有客户端
     **/
    @Override
    public void sendMessageToAllClient(String msg) {
        SseCacheUtil.sendMessageToAllClient(msg);
    }

    /**
     * 根据clientId发送消息给某一客户端
     **/
    @Override
    public void sendMessageToOneClient(String clientId, String msg) {
        SseCacheUtil.sendMessageToOneClient(clientId, msg);
    }

    @Override
    public SseEmitter chat(CommonSseParam param) {
        SseEmitter emitter = this.createSseConnect(param.getClientId(), param.getLoginId(), false);

        HEARTBEAT_EXECUTORS.schedule(
                () -> {
                    String re = "re:" + param.getContent() + System.currentTimeMillis();
                    for (char c : re.toCharArray()) {
                        SseEmitter.SseEventBuilder sendData = SseEmitter.event().id(param.getClientId()).data(Res.ok(c), MediaType.APPLICATION_JSON);
                        try {
                            TimeUnit.MILLISECONDS.sleep(100);
                            emitter.send(sendData);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    emitter.complete();
                }, 3, TimeUnit.SECONDS);


        return emitter;
    }
}
